/*
 * 文件名： NettyChannelManager.java
 *
 * 创建日期： 2017年1月6日
 *
 * Copyright(C) 2017, by <a href="mailto:liws@xingyuanauto.com">liws</a>.
 *
 * 原始作者: liws
 *
 */
package com.moli.iov.gateway.server;

import com.moli.iov.cache.RedisUtil;
import com.moli.iov.gateway.server.model.KafkaSendData;
import com.moli.iov.gateway.server.task.KafkaSendTask;
import com.moli.iov.gateway.util.VinRouter;
import com.moli.iov.protocol.model.dto.VehicleOnLineDto;
import com.moli.iov.util.PropertiesUtils;
import com.moli.iov.util.SpringContextUtils;
import com.moli.iov.util.StringUtil;
import io.netty.channel.Channel;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Server端 Channel管理器
 *
 * @author <a href="mailto:liws@xingyuanauto.com">liws</a>
 * @version $Revision$
 * @since 2017年1月6日
 */
public class NettyChannelManager implements NettyChannelManagerMXBean {


    private static final Logger logger = LoggerFactory.getLogger(NettyChannelManager.class);

    /**
     * redis存储车辆在线的key
     */
    public static final String ONLINE_VEHICLES_KEY = "IOV:VEHICLE:ONLINE";

    private VinThreadPoolExecutor kafkaSendThreadPool;

    private RedisUtil redisUtil;

    private VinRouter vinRouter;

    /***
     * 车辆上线topic
     */
    public static final String VEHICLE_ONLINE = PropertiesUtils.getProperty("server.topic.vehicle.online");
    /**
     * 存放vin与ChannelId对应关系,处于在线状态的（即已登陆的车辆）
     */
    private final ConcurrentHashMap<String, String> vinChannelIdMap = new ConcurrentHashMap<>(20000);

    /**
     * 存放vin与ChannelId对应关系,处于注册状态的,区分目的是防止注册完成后不发登陆直接发实时等异常状态
     */

    private final ConcurrentHashMap<String, String> vinRegisterChannelIdMap = new ConcurrentHashMap<>(20000);

    /**
     * 存放ChannelId与Channel对应关系的,包含登陆和注册使用的channel
     */
    private final ConcurrentHashMap<String, Channel> activeChannelMap = new ConcurrentHashMap<>(20000);

    private NettyChannelManager() {
        redisUtil = SpringContextUtils.getBean("redisUtil");
        kafkaSendThreadPool = SpringContextUtils.getBean("kafkaSendThreadPool");
        vinRouter = SpringContextUtils.getBean("vinRouter");
        removeAllOnlineVehicle();
        // 添加虚拟机钩子 删除redis中 本服务器的在线车辆表
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {

            @Override
            public void run() {
                logger.info("启动钩子函数，清除在线车辆redis");
                removeAllOnlineVehicle();

            }
        }));
    }

    public static NettyChannelManager getInstance() {
        return NettyChannelManagerHolder.instance;
    }

    private static class NettyChannelManagerHolder {

        private static NettyChannelManager instance = new NettyChannelManager();

        private NettyChannelManagerHolder() {
        }
    }

    /**
     * 绑定 Uid与ChannelId的关联关系
     *
     * @param vin
     */
    public synchronized void bindVinAndChannel(String vin, Channel channel) {
        String channelId = channel.id().asLongText();
        if (vin == null || "".equals(vin)){
            return;
        }
        if (channelId == null || "".equals(channelId)) {
            return;
        }

        vinChannelIdMap.put(vin, channelId);
        activeChannelMap.put(channelId, channel);
        logger.info("车辆:" + vin + "上线,本节点车辆在线数:" + vinChannelIdMap.size());
        //向kafka发送上线消息，用于在线率统计和在线状态处理
        VehicleOnLineDto vehicleOnLineDto = new VehicleOnLineDto(vin, true, System.currentTimeMillis());
        kafkaSendThreadPool.execute(vin, new KafkaSendTask(new KafkaSendData(VEHICLE_ONLINE, vin, vehicleOnLineDto)));
    }

    public synchronized void bindUidAndRegisterChannel(String vin, Channel channel) {
        String channelId = channel.id().asLongText();
        if (vin == null || "".equals(vin)) {
            return;
        }
        if (channelId == null || "".equals(channelId)) {
            return;
        }
        vinRegisterChannelIdMap.put(vin, channelId);
        activeChannelMap.put(channelId, channel);
    }

    /**
     * 根据Uid获取Channel
     *
     * @param uid
     */
    public Channel getLoginChannelByVin(String uid) {

        if (uid == null || "".equals(uid)) {
            return null;
        }
        String channelId = vinChannelIdMap.get(uid);
        if (channelId == null || "".equals(channelId)) {
            return null;
        }
        return activeChannelMap.get(channelId);
    }

    public Channel getRegisterChannelByVin(String uid) {

        if (uid == null || "".equals(uid)) {
            return null;
        }
        String channelId = vinRegisterChannelIdMap.get(uid);
        if (channelId == null || "".equals(channelId)) {
            return null;
        }
        return activeChannelMap.get(channelId);
    }

    /**
     * 根据channelID获取已经的登陆车辆的vin
     *
     * @param channelId
     * @return
     */
    public String getLonginVinByChannel(String channelId) {
        String vin = "";
        if (vinChannelIdMap == null || vinChannelIdMap.size() == 0) {
            return null;
        }
        for (Map.Entry<String, String> entry : vinChannelIdMap.entrySet()) {
            if (entry.getValue().equals(channelId)) {
                vin = entry.getKey();
                break;
            }
        }
        return vin;
    }

    /**
     * 根据channelID获取已经的登陆车辆的vin
     *
     * @param channelId
     * @return
     */
    public String getRegisterVinByChannel(String channelId) {
        String vin = "";
        if (vinRegisterChannelIdMap == null || vinRegisterChannelIdMap.size() == 0) {
            return null;
        }
        for (Map.Entry<String, String> entry : vinRegisterChannelIdMap.entrySet()) {
            if (entry.getValue().equals(channelId)) {
                vin = entry.getKey();
                break;
            }
        }
        return vin;
    }

    /**
     * 根据channelId查询对应的vin,可能是从登陆后的channel中获取，也可能是正在注册的channel
     * 都没有返回null，调用方需要验证null
     *
     * @param channelId
     * @return
     */
    public String getVinByChannel(String channelId) {
        try{
            String longinVin = getLonginVinByChannel(channelId);
            if (StringUtil.isNotBlank(longinVin)) {
                return longinVin;
            }
            String registerVin = getRegisterVinByChannel(channelId);
            if (StringUtil.isNotBlank(registerVin)) {
                return registerVin;
            }
        }catch(Exception e){
            logger.error("根据channelId查询对应的vin 异常", e);
        }
        return null;
    }

    /***
     * 根据channelId删除通信通道，断开通道(或并下线车辆)
     * @param channel 通道对象
     */
    public synchronized void removeChannel(Channel channel, String vin) {

        String channelId = channel.id().asLongText();
        logger.info("移除channel中，channelId: {}", channelId);

        //移除 activeChannelMap 中 channel
        removeActiveChannelMap(channelId);

        //移除登陆 ChannelId
        removeVinChannelIdMap(channelId);

        //发送离线Msg
        if(StringUtil.isNotBlank(vin) && vinRouter.isLocalID(vin)){
            offLineSendMsg(vin);
        }

        //3、注册通道中是否有该channel，有删除通道，并直接断开
        removeRegisterChannel(channel);
    }

    /**
     * 移除 activeChannelMap
     * @param channelId
     */
    private boolean removeActiveChannelMap(String channelId){
        try{
            Channel channelOld = this.activeChannelMap.remove(channelId);
            if (channelOld == null) {
                return true;
            }
        }catch(Exception e){
            logger.error("移除 activeChannelMap", e);
        }
        logger.error("移除 activeChannelMap中channelr失败 channelId: {}", channelId);
        return false;
    }

    /**
     * 移除 登陆  VinChannelIdMap 中的channelId
     * @param channeId
     */
    private boolean removeVinChannelIdMap(String channeId){
        try {
            Iterator<Entry<String, String>> iterator = vinChannelIdMap.entrySet().iterator();
            while(iterator.hasNext()){
                Entry<String, String> next = iterator.next();
                if(next.getValue().equals(channeId)){
                    iterator.remove();
                    return true;
                }
            }
        }catch(Exception e){
            logger.error("移除 登陆  VinChannelIdMap 中的channelId", e);
        }
        return false;
    }

    /**
     * 移除
     * 移除vin与ChannelId对应关系
     * 并发送离线消息
     * @param vin
     */
    private void offLineSendMsg(String vin){
        try{
            VehicleOnLineDto vehicleOnLineDto = new VehicleOnLineDto(vin, false, System.currentTimeMillis());
            kafkaSendThreadPool.execute(vin, new KafkaSendTask(new KafkaSendData(VEHICLE_ONLINE, vin, vehicleOnLineDto)));
            logger.info("车辆:" + vin + "下线,本节点车辆在线数:" + vinChannelIdMap.size());
        }catch(Exception e){
            logger.error("离线 offLineSendMsg 异常", e);
        }
    }

    /**
     * 根据channelId删除注册通道中的channel
     *
     * @param channel
     */
    public void removeRegisterChannel(Channel channel) {
        if (channel == null) {
            return;
        }
        String channelId = channel.id().asLongText();
        for (Map.Entry<String, String> entry : vinRegisterChannelIdMap.entrySet()) {
            String key = entry.getKey();
            String val = entry.getValue();
            if (channelId.equals(val)) {
                vinRegisterChannelIdMap.remove(key);
                break;
            }
        }
    }

     /**
     * 删除所有此节点上所有车辆在线信息
     */
    public void removeAllOnlineVehicle() {

        Map<String, String> map = getVinChannelMap();
        Set<String> keySet = new HashSet<>();
        for (Map.Entry<String, String> entry : map.entrySet()) {
            keySet.add(entry.getKey());
        }
        String[] keyArray = new String[keySet.size()];
        keySet.toArray(keyArray);
        logger.info("[系统关闭Del在线车辆][Key=" + ONLINE_VEHICLES_KEY + "]");
        if (keyArray.length <= 0) {
            return;
        }
        try {
            redisUtil.hdel(ONLINE_VEHICLES_KEY, keyArray);
            vinChannelIdMap.clear();
            vinRegisterChannelIdMap.clear();
            activeChannelMap.clear();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
        }
    }

    /**
     * 获取redis中vin与channel所有集合
     *
     * @return
     */
    public Map<String, String> getVinChannelMap() {
        Map<String, String> map = null;
        try {
            map = redisUtil.hgetString(ONLINE_VEHICLES_KEY);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        }
        return map;
    }

    /**
     * 添加通道
     *
     * @param channel
     */
    public void putChannel(Channel channel) {
        this.activeChannelMap.put(channel.id().asLongText(), channel);
    }

    /**
     * @see
     */
    @Override
    public long getUidAndChannelSize() {
        return vinChannelIdMap.size();
    }

    /**
     * @see
     */
    @Override
    public long getActiveChannelSize() {
        return activeChannelMap.size();
    }


    public ConcurrentHashMap<String, String> getActiveVinChannelMap() {
        return vinChannelIdMap;
    }

    /**
     * @see
     */
    @Override
    public String getUidChannelIdText() {
        StringBuilder sBuilder = new StringBuilder();
        for (Entry<String, String> entry : vinChannelIdMap.entrySet()) {
            sBuilder.append("[");
            sBuilder.append(entry.getKey());
            sBuilder.append("<->");
            sBuilder.append(entry.getValue());
            Channel channel = activeChannelMap.get(entry.getValue());
            if (channel != null) {
                sBuilder.append("{");
                sBuilder.append(channel.remoteAddress().toString());
                sBuilder.append("}");
            }
            sBuilder.append("]");
        }
        return sBuilder.toString();
    }

}
